package com.zjc.springboot.storm.bolt;

import com.zjc.springboot.storm.messages.StormMsg;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class KafkaBolt implements IRichBolt {

    private static Logger logger = LoggerFactory.getLogger(KafkaBolt.class);

    protected OutputCollector collector;

    @Override
    public void prepare(Map map, TopologyContext context, OutputCollector collector) {

        this.collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        try {
            Object obj = tuple.getValue(0);
            if (obj instanceof StormMsg) {
                StormMsg stormMsg = (StormMsg) obj;
                if (stormMsg != null) {
                    logger.info("bolt tuple.value : " + stormMsg);
                    this.collector.ack(tuple);
                }
            }
        } catch (Exception e) {
            logger.error("Data Process Error.", e);
            this.collector.fail(tuple);
        }
    }

    @Override
    public void cleanup() {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(new String[]{"uniqueNo", "alarm"}));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
